package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 */
public class Demo14_OverWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            //在流中产生水印
            .assignTimestampsAndWatermarks(watermarkStrategy);

        Table table = tableEnv.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),
            $("et").rowtime());

        tableEnv.createTemporaryView("t1",table);

        String sql1 = "select id,ts, sum(vc) over(partition by id order by pt rows between unbounded preceding and current row ) sumVc  from t1 ";
        String sql2 = "select id,ts, sum(vc) over(partition by id order by pt rows between 2 preceding and current row ) sumVc  from t1 ";

        String sql3 = "select id,ts, sum(vc) over(partition by id order by et range between unbounded preceding and current row ) sumVc  from t1 ";
        String sql4 = "select id,ts, sum(vc) over(partition by id order by et range between INTERVAL '2' SECOND preceding and current row ) sumVc  from t1 ";

        tableEnv.sqlQuery(sql4)
                        .execute()
                        .print();

    }
}
